package tom.reactor.service;

import com.fasterxml.jackson.core.io.JsonStringEncoder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import tom.reactor.model.MsgEvent;

/**
 * @author ZHUFEIFEI
 */
@Slf4j
@Service
public class MsgEventService {

    @Autowired
    private ReactiveRedisTemplate reactiveRedisTemplate;

    private static final String EVENT_KEY = "event";

    public Mono<Void> saveEvent(Flux<MsgEvent> events) {
        return events.flatMap(m -> {
            this.reactiveRedisTemplate.opsForList()
                    .leftPush(EVENT_KEY, m)
                    .doOnSuccess((val) -> log.info("lpush : {} success!", val))
                    .doOnError(exp -> {log.error("lpush : {} error : {}",m, exp);});
            return null;
        }).log().then();
    }

    public Flux<MsgEvent> findEvent() {
        return this.reactiveRedisTemplate.opsForList().rightPop(EVENT_KEY)
                .flux();
    }


}
